"""
Excel数据处理服务
支持Excel文件的上传、解析和数据验证
"""
import os
import pandas as pd
import numpy as np
from typing import Dict, Any, List, Optional, Union, Tuple
import json
from datetime import datetime
import tempfile
import shutil
from pathlib import Path

from core.config import settings
from core.logging_config import get_logger
from core.exceptions import DatabaseError, ValidationError

logger = get_logger("excel_data_processor")


class ExcelProcessor:
    """Excel文件处理器"""
    
    def __init__(self):
        self.supported_formats = ['.xlsx', '.xls', '.csv']
        self.max_file_size = settings.max_file_size
        self.upload_dir = Path(settings.upload_dir)
        self.upload_dir.mkdir(exist_ok=True)
    
    def validate_file(self, file_path: str) -> Dict[str, Any]:
        """验证Excel文件"""
        try:
            path = Path(file_path)
            
            # 检查文件是否存在
            if not path.exists():
                return {"valid": False, "message": "文件不存在"}
            
            # 检查文件扩展名
            if path.suffix.lower() not in self.supported_formats:
                return {
                    "valid": False, 
                    "message": f"不支持的文件格式，支持: {', '.join(self.supported_formats)}"
                }
            
            # 检查文件大小
            file_size = path.stat().st_size
            if file_size > self.max_file_size:
                return {
                    "valid": False,
                    "message": f"文件大小超过限制 (最大: {self.max_file_size // 1024 // 1024}MB)"
                }
            
            # 尝试读取文件
            try:
                if path.suffix.lower() == '.csv':
                    df = pd.read_csv(file_path, nrows=1)
                else:
                    df = pd.read_excel(file_path, nrows=1)
            except Exception as e:
                return {"valid": False, "message": f"文件读取失败: {str(e)}"}
            
            return {
                "valid": True,
                "message": "文件验证通过",
                "file_size": file_size,
                "file_format": path.suffix.lower(),
                "file_name": path.name
            }
            
        except Exception as e:
            logger.error(f"文件验证失败: {str(e)}")
            return {"valid": False, "message": f"验证失败: {str(e)}"}
    
    def read_excel_file(self, file_path: str, sheet_name: Union[str, int, List[str], None] = 0) -> Dict[str, Any]:
        """读取Excel文件"""
        try:
            validation_result = self.validate_file(file_path)
            if not validation_result["valid"]:
                raise ValidationError(validation_result["message"])
            
            file_format = validation_result["file_format"]
            
            # 读取Excel文件
            if file_format == '.csv':
                df = pd.read_csv(file_path, encoding='utf-8')
                sheets_data = {"Sheet1": df}
            else:
                # 读取所有工作表或指定工作表
                if sheet_name is None:
                    sheets_data = pd.read_excel(file_path, sheet_name=None)
                else:
                    df = pd.read_excel(file_path, sheet_name=sheet_name)
                    if isinstance(sheet_name, str):
                        sheets_data = {sheet_name: df}
                    else:
                        sheets_data = {"Sheet1": df}
            
            # 处理每个工作表
            result = {}
            for sheet_name, df in sheets_data.items():
                sheet_info = self._process_dataframe(df, sheet_name)
                result[sheet_name] = sheet_info
            
            return {
                "success": True,
                "data": result,
                "file_info": validation_result
            }
            
        except Exception as e:
            logger.error(f"读取Excel文件失败: {str(e)}")
            return {
                "success": False,
                "message": f"读取失败: {str(e)}",
                "data": {},
                "file_info": {}
            }
    
    def _process_dataframe(self, df: pd.DataFrame, sheet_name: str) -> Dict[str, Any]:
        """处理DataFrame数据"""
        try:
            # 基本信息
            row_count = len(df)
            column_count = len(df.columns)
            
            # 列信息
            columns_info = []
            for col in df.columns:
                col_data = df[col]
                columns_info.append({
                    "name": str(col),
                    "type": str(col_data.dtype),
                    "null_count": col_data.isnull().sum(),
                    "unique_count": col_data.nunique(),
                    "sample_values": col_data.dropna().head(3).tolist()
                })
            
            # 数据质量分析
            quality_analysis = self._analyze_data_quality(df)
            
            # 数据预览（前10行）
            preview_data = df.head(10).fillna("").to_dict('records')
            
            return {
                "sheet_name": sheet_name,
                "row_count": row_count,
                "column_count": column_count,
                "columns": columns_info,
                "quality_analysis": quality_analysis,
                "preview_data": preview_data,
                "data_types": self._infer_data_types(df)
            }
            
        except Exception as e:
            logger.error(f"处理DataFrame失败: {str(e)}")
            raise DatabaseError(f"处理数据失败: {str(e)}")
    
    def _analyze_data_quality(self, df: pd.DataFrame) -> Dict[str, Any]:
        """分析数据质量"""
        try:
            total_rows = len(df)
            total_columns = len(df.columns)
            
            # 计算各种质量指标
            null_count = df.isnull().sum().sum()
            duplicate_count = df.duplicated().sum()
            
            # 列质量分析
            column_quality = []
            for col in df.columns:
                col_data = df[col]
                null_percentage = (col_data.isnull().sum() / total_rows) * 100 if total_rows > 0 else 0
                
                column_quality.append({
                    "column_name": str(col),
                    "null_count": int(col_data.isnull().sum()),
                    "null_percentage": round(null_percentage, 2),
                    "unique_count": int(col_data.nunique()),
                    "duplicate_count": int(col_data.duplicated().sum()),
                    "data_type": str(col_data.dtype)
                })
            
            return {
                "total_rows": total_rows,
                "total_columns": total_columns,
                "total_null_count": int(null_count),
                "total_duplicate_count": int(duplicate_count),
                "null_percentage": round((null_count / (total_rows * total_columns)) * 100, 2) if total_rows > 0 else 0,
                "duplicate_percentage": round((duplicate_count / total_rows) * 100, 2) if total_rows > 0 else 0,
                "column_quality": column_quality,
                "quality_score": self._calculate_quality_score(df)
            }
            
        except Exception as e:
            logger.error(f"数据质量分析失败: {str(e)}")
            return {}
    
    def _calculate_quality_score(self, df: pd.DataFrame) -> float:
        """计算数据质量分数"""
        try:
            total_cells = len(df) * len(df.columns)
            if total_cells == 0:
                return 0.0
            
            null_cells = df.isnull().sum().sum()
            duplicate_rows = df.duplicated().sum()
            
            # 计算分数 (0-100)
            quality_score = 100.0
            
            # 扣除缺失值分数
            null_percentage = (null_cells / total_cells) * 100
            quality_score -= null_percentage * 0.5
            
            # 扣除重复行分数
            duplicate_percentage = (duplicate_rows / len(df)) * 100 if len(df) > 0 else 0
            quality_score -= duplicate_percentage * 0.3
            
            return max(0.0, min(100.0, quality_score))
            
        except Exception as e:
            logger.error(f"计算质量分数失败: {str(e)}")
            return 0.0
    
    def _infer_data_types(self, df: pd.DataFrame) -> Dict[str, str]:
        """推断数据类型"""
        try:
            data_types = {}
            for col in df.columns:
                col_data = df[col].dropna()
                
                if len(col_data) == 0:
                    data_types[str(col)] = "empty"
                    continue
                
                # 尝试推断类型
                sample_data = col_data.head(100)
                
                # 检查是否为数值型
                try:
                    pd.to_numeric(sample_data, errors='raise')
                    data_types[str(col)] = "numeric"
                    continue
                except:
                    pass
                
                # 检查是否为日期型
                try:
                    pd.to_datetime(sample_data, errors='raise')
                    data_types[str(col)] = "datetime"
                    continue
                except:
                    pass
                
                # 检查是否为布尔型
                if sample_data.dtype == 'bool':
                    data_types[str(col)] = "boolean"
                    continue
                
                # 检查是否为类别型
                unique_ratio = len(sample_data.unique()) / len(sample_data)
                if unique_ratio < 0.1:
                    data_types[str(col)] = "categorical"
                else:
                    data_types[str(col)] = "text"
            
            return data_types
            
        except Exception as e:
            logger.error(f"推断数据类型失败: {str(e)}")
            return {}
    
    def clean_data(self, df: pd.DataFrame, cleaning_config: Dict[str, Any]) -> pd.DataFrame:
        """清洗数据"""
        try:
            cleaned_df = df.copy()
            
            # 处理缺失值
            if cleaning_config.get("handle_nulls"):
                method = cleaning_config.get("null_method", "drop")
                if method == "drop":
                    cleaned_df = cleaned_df.dropna()
                elif method == "fill_mean":
                    numeric_cols = cleaned_df.select_dtypes(include=[np.number]).columns
                    cleaned_df[numeric_cols] = cleaned_df[numeric_cols].fillna(cleaned_df[numeric_cols].mean())
                elif method == "fill_median":
                    numeric_cols = cleaned_df.select_dtypes(include=[np.number]).columns
                    cleaned_df[numeric_cols] = cleaned_df[numeric_cols].fillna(cleaned_df[numeric_cols].median())
                elif method == "fill_mode":
                    for col in cleaned_df.columns:
                        mode_value = cleaned_df[col].mode()
                        if len(mode_value) > 0:
                            cleaned_df[col] = cleaned_df[col].fillna(mode_value[0])
            
            # 删除重复行
            if cleaning_config.get("remove_duplicates", True):
                cleaned_df = cleaned_df.drop_duplicates()
            
            # 数据类型转换
            if cleaning_config.get("convert_types"):
                type_mapping = cleaning_config.get("type_mapping", {})
                for col, target_type in type_mapping.items():
                    if col in cleaned_df.columns:
                        try:
                            if target_type == "numeric":
                                cleaned_df[col] = pd.to_numeric(cleaned_df[col], errors='coerce')
                            elif target_type == "datetime":
                                cleaned_df[col] = pd.to_datetime(cleaned_df[col], errors='coerce')
                            elif target_type == "string":
                                cleaned_df[col] = cleaned_df[col].astype(str)
                        except Exception as e:
                            logger.warning(f"列 {col} 类型转换失败: {str(e)}")
            
            return cleaned_df
            
        except Exception as e:
            logger.error(f"数据清洗失败: {str(e)}")
            raise DatabaseError(f"数据清洗失败: {str(e)}")
    
    def export_to_dataframe(self, file_path: str, sheet_name: Union[str, int, None] = 0, 
                          cleaning_config: Optional[Dict[str, Any]] = None) -> pd.DataFrame:
        """导出为DataFrame"""
        try:
            result = self.read_excel_file(file_path, sheet_name)
            
            if not result["success"]:
                raise DatabaseError(result["message"])
            
            # 获取第一个工作表的数据
            sheets_data = result["data"]
            if not sheets_data:
                raise DatabaseError("没有找到数据")
            
            sheet_key = list(sheets_data.keys())[0]
            sheet_info = sheets_data[sheet_key]
            
            # 重新读取完整数据
            if Path(file_path).suffix.lower() == '.csv':
                df = pd.read_csv(file_path, encoding='utf-8')
            else:
                df = pd.read_excel(file_path, sheet_name=sheet_name if sheet_name else 0)
            
            # 数据清洗
            if cleaning_config:
                df = self.clean_data(df, cleaning_config)
            
            return df
            
        except Exception as e:
            logger.error(f"导出DataFrame失败: {str(e)}")
            raise DatabaseError(f"导出失败: {str(e)}")
    
    def save_uploaded_file(self, file_content: bytes, original_filename: str) -> str:
        """保存上传的文件"""
        try:
            # 生成安全的文件名
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            file_extension = Path(original_filename).suffix
            safe_filename = f"{timestamp}_{original_filename}"
            
            # 保存文件
            file_path = self.upload_dir / safe_filename
            with open(file_path, 'wb') as f:
                f.write(file_content)
            
            logger.info(f"文件保存成功: {safe_filename}")
            return str(file_path)
            
        except Exception as e:
            logger.error(f"文件保存失败: {str(e)}")
            raise DatabaseError(f"文件保存失败: {str(e)}")
    
    def delete_file(self, file_path: str):
        """删除文件"""
        try:
            path = Path(file_path)
            if path.exists():
                path.unlink()
                logger.info(f"文件删除成功: {file_path}")
            else:
                logger.warning(f"文件不存在: {file_path}")
                
        except Exception as e:
            logger.error(f"文件删除失败: {str(e)}")
            raise DatabaseError(f"文件删除失败: {str(e)}")


class ExcelTemplate:
    """Excel模板管理"""
    
    def __init__(self):
        self.templates = self._load_templates()
    
    def _load_templates(self) -> Dict[str, Any]:
        """加载预定义模板"""
        return {
            "petition_data": {
                "name": "信访数据模板",
                "description": "标准的信访数据导入模板",
                "columns": [
                    {"name": "petition_id", "type": "string", "required": True, "description": "信访ID"},
                    {"name": "title", "type": "string", "required": True, "description": "标题"},
                    {"name": "content", "type": "string", "required": True, "description": "内容"},
                    {"name": "submit_time", "type": "datetime", "required": True, "description": "提交时间"},
                    {"name": "complainant_name", "type": "string", "required": False, "description": "投诉人姓名"},
                    {"name": "respondent_name", "type": "string", "required": False, "description": "被投诉方"},
                    {"name": "province", "type": "string", "required": False, "description": "省份"},
                    {"name": "city", "type": "string", "required": False, "description": "城市"},
                    {"name": "pollution_type", "type": "string", "required": False, "description": "污染类型"}
                ]
            },
            "enterprise_data": {
                "name": "企业数据模板",
                "description": "企业信息数据导入模板",
                "columns": [
                    {"name": "enterprise_id", "type": "string", "required": True, "description": "企业ID"},
                    {"name": "enterprise_name", "type": "string", "required": True, "description": "企业名称"},
                    {"name": "industry_type", "type": "string", "required": True, "description": "行业类型"},
                    {"name": "address", "type": "string", "required": True, "description": "地址"},
                    {"name": "latitude", "type": "numeric", "required": False, "description": "纬度"},
                    {"name": "longitude", "type": "numeric", "required": False, "description": "经度"}
                ]
            }
        }
    
    def get_template(self, template_name: str) -> Optional[Dict[str, Any]]:
        """获取模板"""
        return self.templates.get(template_name)
    
    def get_all_templates(self) -> Dict[str, Any]:
        """获取所有模板"""
        return self.templates
    
    def validate_against_template(self, df: pd.DataFrame, template_name: str) -> Dict[str, Any]:
        """根据模板验证数据"""
        template = self.get_template(template_name)
        if not template:
            return {"valid": False, "message": f"模板 {template_name} 不存在"}
        
        try:
            required_columns = [col["name"] for col in template["columns"] if col.get("required", False)]
            template_columns = [col["name"] for col in template["columns"]]
            
            df_columns = list(df.columns)
            
            # 检查必需列
            missing_columns = [col for col in required_columns if col not in df_columns]
            if missing_columns:
                return {
                    "valid": False,
                    "message": f"缺少必需列: {', '.join(missing_columns)}",
                    "missing_columns": missing_columns
                }
            
            # 检查列类型匹配
            type_mismatches = []
            for col in template_columns:
                if col in df_columns:
                    template_type = next((c["type"] for c in template["columns"] if c["name"] == col), None)
                    if template_type:
                        actual_type = str(df[col].dtype)
                        if not self._is_type_compatible(actual_type, template_type):
                            type_mismatches.append({
                                "column": col,
                                "expected": template_type,
                                "actual": actual_type
                            })
            
            return {
                "valid": len(type_mismatches) == 0,
                "message": "验证通过" if len(type_mismatches) == 0 else f"类型不匹配: {len(type_mismatches)} 列",
                "type_mismatches": type_mismatches
            }
            
        except Exception as e:
            logger.error(f"模板验证失败: {str(e)}")
            return {"valid": False, "message": f"验证失败: {str(e)}"}
    
    def _is_type_compatible(self, actual_type: str, expected_type: str) -> bool:
        """检查类型是否兼容"""
        type_mapping = {
            "numeric": ["int64", "float64", "int32", "float32"],
            "string": ["object", "string"],
            "datetime": ["datetime64", "object"],
            "boolean": ["bool"]
        }
        
        expected_types = type_mapping.get(expected_type, [])
        return any(actual_type.startswith(t) for t in expected_types)


# 全局Excel处理器
excel_processor = ExcelProcessor()
excel_template = ExcelTemplate()